package cn.tedu.charging.device.warm;

import cn.tedu.charging.common.pojo.po.ChargingStationPO;
import cn.tedu.charging.device.repository.StationRepository;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.data.geo.Point;
import org.springframework.data.redis.core.GeoOperations;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;

@Component
@Slf4j
public class RedisWarmUp implements ApplicationRunner {
    @Autowired
    private StationRepository stationRepository;
    @Resource
    @Qualifier("redis01")
    private RedisTemplate redisCache;

    @Value("${cache.warm.enabled}")
    private boolean open;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        // 缓存预热开关，防止集群启动时全部操作缓存
        if (!open) {
            log.debug("缓存预热未开启");
            return;
        }
        //setnx warm:up:lock 11  设置集群锁
        Boolean result = redisCache.boundValueOps("warm:up:lock").setIfAbsent("11");
        if (!result){
            log.debug("抢锁失败,无权执行预热");
            return;
        }
        // 1. 读取全量数据
        // 1.1 读取total 计算批量 （假设是10万条，按照每1000条一次，做100次循环，设计到分页读）
        Long total = stationRepository.getTotal();
        log.debug("查询数据库战场数据总条数:{}", total);
        // 1.2 按批量执行循环 total 比 1000 计算循环次数
        Long loop = total % 1000 == 0 ? total / 1000 : total / 1000 + 1;
        log.debug("循环次数：{}", loop);
        for (int i = 0; i < loop; i++) {
            log.debug("开始处理第{}批缓存预热", i + 1);
            List<ChargingStationPO> pos = stationRepository.getPage(i + 1, 1000);
            // 2. 将每个批量的数据，第一个1000，第二个1000条...做缓存写入
            createCache(pos);
        }

    }

    private void createCache(List<ChargingStationPO> pos) {
        if (pos != null && pos.size() > 0) {
            // 通过 pipeline 批量写入，减少网络请求
            redisCache.executePipelined((RedisCallback<Object>) connection -> {
                //将你所需要执行的所有命令 放到这一次pipeline中
                for (ChargingStationPO po : pos) {
                    //connection是底层连接 需要手动序列化 反序列化 直接操作的api 类似redis命名
                    //connection.geoAdd("beijing:stations".getBytes(StandardCharsets.UTF_8),)
                    saveGeo(po);
                    saveString(po);
                }
                return null;//返回本次执行结果
            });
            for (ChargingStationPO po : pos) {
                saveString(po);
                saveGeo(po);
            }
        }
    }

    private void saveString(ChargingStationPO po) {
        // set stationId stationPO
        String key = "station:detail:" + po.getId();
        redisCache.boundValueOps(key).set(po);
    }

    private void saveGeo(ChargingStationPO po) {
        // geoadd po.getCity po.getlon po.getlat po.getId
        String geoKey = "beijing:stations";
        GeoOperations geoOps = redisCache.opsForGeo();
        geoOps.add(geoKey, new Point(po.getStationLng().doubleValue(), po.getStationLat().doubleValue()), po.getId());
    }
}
